// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: change E2E packet format
// feature: Support Cache fast application and slow release mechanism
// feature: develop dispatcher
// feature: Split history as an independent common module
// feature: destination order support by reception timestamp
// feature: E2E Profile04 and Configuration Function Development
// ------------------------------------------------------------------

#include <edds/rtps/reader/RTPSReader.h>

#include <history/IChangePool.h>
#include <history/BasicPayloadPool.hpp>

#include <rtps/participant/RTPSParticipantImpl.h>

#include <rtps/reader/ReaderHistoryState.hpp>

#include <history/ReaderHistory.h>

#include <history/ReaderListener.h>
#include <deps/resources/ResourceEvent.h>
#include <deps/resources/TimedEvent.h>

#include <typeinfo>
#include <algorithm>
#include <chrono>

#include <history/ReaderHistory.h>
#include <elog/Log.hpp>

#include <foonathan/memory/namespace_alias.hpp>
#include <history/TopicPayloadPoolRegistry.hpp>

#define IDSTRING "(ID:" << std::this_thread::get_id() << ") " <<

using namespace vbs::common;
namespace evbs {
namespace ertps {
namespace rtps {

RTPSReader::RTPSReader(RTPSParticipantImpl* pimpl, const GUID_t& guid, const ReaderAttributes& att, ReaderHistory* hist,
                       vbs::common::ReaderListener* rlisten)
    : Endpoint(pimpl, guid, att.endpoint),
      mp_history(hist),
      mp_listener(rlisten),
      m_acceptMessagesToUnknownReaders(true),
      m_acceptMessagesFromUnkownWriters(false),
      history_state_(new ReaderHistoryState(att.matched_writers_allocation.initial)),
      liveliness_kind_(att.liveliness_kind_),
      liveliness_lease_duration_(att.liveliness_lease_duration),
      enable_e2e_protection_(att.enable_e2e_protection),
      e2e_p04_min_data_length_(att.e2e_p04_min_data_length),
      e2e_p04_max_data_length_(att.e2e_p04_max_data_length),
      e2e_p04_max_delta_counter_(att.e2e_p04_max_delta_counter),
      trace_point_status_(att.trace_point_status) {
    PoolConfig cfg = PoolConfig::from_history_attributes(hist->m_att);
    std::shared_ptr<IChangePool> change_pool;
    std::shared_ptr<IPayloadPool> payload_pool;
    payload_pool = BasicPayloadPool::get(cfg, change_pool);
    //mp_history->change_pool_ = change_pool;

    init(payload_pool, change_pool, att);
}

RTPSReader::RTPSReader(RTPSParticipantImpl* pimpl, const GUID_t& guid, const ReaderAttributes& att,
                       const std::shared_ptr<IPayloadPool>& payload_pool, ReaderHistory* hist,
                       vbs::common::ReaderListener* rlisten)
    : RTPSReader(pimpl, guid, att, payload_pool,
                 std::make_shared<CacheChangePool>(PoolConfig::from_history_attributes(hist->m_att)), hist, rlisten) {}

RTPSReader::RTPSReader(RTPSParticipantImpl* pimpl, const GUID_t& guid, const ReaderAttributes& att,
                       const std::shared_ptr<IChangePool>& change_pool, ReaderHistory* hist,
                       vbs::common::ReaderListener* rlisten)
    : RTPSReader(pimpl, guid, att, get_payload_pool(hist), change_pool, hist, rlisten) {}

RTPSReader::RTPSReader(RTPSParticipantImpl* pimpl, const GUID_t& guid, const ReaderAttributes& att,
                       const std::shared_ptr<IPayloadPool>& payload_pool,
                       const std::shared_ptr<IChangePool>& change_pool, ReaderHistory* hist,
                       vbs::common::ReaderListener* rlisten)
    : Endpoint(pimpl, guid, att.endpoint),
      mp_history(hist),
      mp_listener(rlisten),
      m_acceptMessagesToUnknownReaders(true),
      m_acceptMessagesFromUnkownWriters(false),
      history_state_(new ReaderHistoryState(att.matched_writers_allocation.initial)),
      liveliness_kind_(att.liveliness_kind_),
      liveliness_lease_duration_(att.liveliness_lease_duration),
      enable_e2e_protection_(att.enable_e2e_protection),
      e2e_p04_min_data_length_(att.e2e_p04_min_data_length),
      e2e_p04_max_data_length_(att.e2e_p04_max_data_length),
      e2e_p04_max_delta_counter_(att.e2e_p04_max_delta_counter),
      trace_point_status_(att.trace_point_status) {
    init(payload_pool, change_pool, att);
}

void RTPSReader::init(const std::shared_ptr<IPayloadPool>& payload_pool,
                      const std::shared_ptr<IChangePool>& change_pool, const ReaderAttributes& att) {
    UNUSED_PARAMETER(att);
    payload_pool_ = payload_pool;
    change_pool_ = change_pool;
    fixed_payload_size_ = 0U;
    if (mp_history->m_att.memoryPolicy == PREALLOCATED_MEMORY_MODE) {
        fixed_payload_size_ = mp_history->m_att.payloadMaxSize;
    }

    //mp_history->mp_reader = this;
    //mp_history->mp_mutex  = &mp_mutex;
    //mp_mutex = &(mp_history->mp_mutex);
    mp_history->dest_order_reception = att.dest_order_reception;
    if (mp_history->change_pool_ == nullptr) {
        mp_history->change_pool_ = change_pool;
    }
    if (mp_history->payload_pool_ == nullptr) {
        mp_history->payload_pool_ = payload_pool;
    }

    if ((!this->m_guid.is_builtin()) && mp_history->autoMemoryMode) {
        mem_watermark = mp_history->m_att.maximumReservedCaches > 0 ? mp_history->m_att.maximumReservedCaches : 5000;
        free_event_ = new TimedEvent(
            getRTPSParticipant()->getEventResource(), [&]() -> bool { return history_adjustment(); }, 30000);
        free_event_->restart_timer();
    } else {
        free_event_ = nullptr;
    }
    logDebug(RTPS_READER, "RTPSReader " << this->getGuid() << " created correctly");
}

RTPSReader::~RTPSReader() {
    logDebug(RTPS_READER, "Removing reader " << this->getGuid());
    setListener(nullptr);
    if (free_event_ != nullptr) {
        delete (free_event_);
        free_event_ = nullptr;
    }
#if 0
    for (auto it = mp_history->changesBegin(); it != mp_history->changesEnd(); ++it) {
        releaseCache(*it);
    }
#endif
    if (!m_guid.is_builtin() && payload_pool_ != nullptr) {
        release_payload_pool();
    }

    delete history_state_;
    //mp_history->mp_reader = nullptr;
    //mp_history->mp_mutex  = nullptr;
}

bool RTPSReader::reserveCache(CacheChange_t** change, uint32_t dataCdrSerializedSize) {
    std::lock_guard<RecursiveTimedMutex> guard(mp_history->getMutex());

    *change = nullptr;

    CacheChange_t* reserved_change = nullptr;
    if (!change_pool_->reserve_cache(reserved_change)) {
        logWarning(RTPS_READER, "Problem reserving cache from pool");
        return false;
    }

    uint32_t payload_size = fixed_payload_size_ ? fixed_payload_size_ : dataCdrSerializedSize;
    if (!payload_pool_->get_payload(payload_size, *reserved_change)) {
        (void)change_pool_->release_cache(reserved_change);
        logWarning(RTPS_READER, "Problem reserving payload from pool");
        return false;
    }

    *change = reserved_change;
    return true;
}

void RTPSReader::releaseCache(CacheChange_t* change) {
    std::lock_guard<RecursiveTimedMutex> guard(mp_history->getMutex());

    IPayloadPool* pool = change->payload_owner();
    if (pool) {
        (void)pool->release_payload(*change);
    }
    (void)change_pool_->release_cache(change);
}

vbs::common::ReaderListener* RTPSReader::getListener() const {
    return mp_listener;
}

bool RTPSReader::setListener(vbs::common::ReaderListener* target) {
    std::lock_guard<RecursiveTimedMutex> guard(mp_mutex);
    mp_listener = target;
    return true;
}

History::const_iterator RTPSReader::findCacheInFragmentedProcess(const SequenceNumber_t& sequence_number,
                                                                 const GUID_t& writer_guid, CacheChange_t** change,
                                                                 History::const_iterator hint) const {
    History::const_iterator ret_val = mp_history->get_change_nts(sequence_number, writer_guid, change, hint);

    if ((nullptr != *change) && (*change)->is_fully_assembled()) {
        *change = nullptr;
    }

    return ret_val;
}

void RTPSReader::add_persistence_guid(const GUID_t& guid, const GUID_t& persistence_guid) {
    if (c_Guid_Unknown == persistence_guid || persistence_guid == guid) {
        std::lock_guard<RecursiveTimedMutex> guard(mp_mutex);
        try {
            history_state_->persistence_guid_map[guid] = guid;
            history_state_->persistence_guid_count[guid]++;
        } catch (const foonathan::memory::bad_node_size& e) {  //LCOV_EXCL_START
            logWarning(RTPS_READER, "same guid " << guid << " for persistence guid " << persistence_guid
                                                 << "catch error: " << e.what());
        }  //LCOV_EXCL_STOP
    } else {
        std::lock_guard<RecursiveTimedMutex> guard(mp_mutex);
        try {
            history_state_->persistence_guid_map[guid] = persistence_guid;
            history_state_->persistence_guid_count[persistence_guid]++;
        } catch (const foonathan::memory::bad_node_size& e) {  //LCOV_EXCL_START
            logWarning(RTPS_READER, "diff guid " << guid << " for persistence guid " << persistence_guid
                                                 << "catch bad_node_size: " << e.what());
        } catch (const foonathan::memory::out_of_memory& e) {
            logWarning(RTPS_READER, "diff guid " << guid << " for persistence guid " << persistence_guid
                                                 << "catch out_of_memory: " << e.what());
        } catch (const std::exception& e) {
            logWarning(RTPS_READER, "diff guid " << guid << " for persistence guid " << persistence_guid
                                                 << "catch other: " << e.what());
        }  //LCOV_EXCL_STOP

        // Could happen that a value has already been stored in the record with the guid and not the
        // persistence guid This is because received_change is called before Proxy is created In
        // this case, we substitute the guid for the persistence (in case they are not equal)
        auto spourious_record = history_state_->history_record.find(guid);
        if (spourious_record != history_state_->history_record.end()) {
            logDebug(RTPS_READER,
                     "Sporious record found, changing guid " << guid << " for persistence guid " << persistence_guid);
            (void)update_last_notified(guid, spourious_record->second);
            (void)history_state_->history_record.erase(spourious_record);
        }
    }
}

bool RTPSReader::may_remove_history_record(bool removed_by_lease) {
    return !removed_by_lease;
}

void RTPSReader::remove_persistence_guid(const GUID_t& guid, const GUID_t& persistence_guid, bool removed_by_lease) {
    std::lock_guard<RecursiveTimedMutex> guard(mp_mutex);
    GUID_t persistence_guid_stored = (c_Guid_Unknown == persistence_guid) ? guid : persistence_guid;
    (void)history_state_->persistence_guid_map.erase(guid);
    auto count = --history_state_->persistence_guid_count[persistence_guid_stored];
    if ((count <= 0U) && may_remove_history_record(removed_by_lease)) {
        (void)history_state_->history_record.erase(persistence_guid_stored);
        (void)history_state_->persistence_guid_count.erase(persistence_guid_stored);
    } else if ((count <= 0U) && (c_Guid_Unknown == persistence_guid)) {
        (void)history_state_->history_record.erase(persistence_guid_stored);
        (void)history_state_->persistence_guid_count.erase(persistence_guid_stored);
    }
}

SequenceNumber_t RTPSReader::update_last_notified(const GUID_t& guid, const SequenceNumber_t& seq) {
    SequenceNumber_t ret_val;
    std::lock_guard<RecursiveTimedMutex> guard(mp_mutex);
    GUID_t guid_to_look = guid;
    auto p_guid = history_state_->persistence_guid_map.find(guid);
    if (p_guid != history_state_->persistence_guid_map.end()) {
        guid_to_look = p_guid->second;
    }

    auto p_seq = history_state_->history_record.find(guid_to_look);
    if (p_seq != history_state_->history_record.end()) {
        ret_val = p_seq->second;
    }

    if (ret_val < seq) {
        set_last_notified(guid_to_look, seq);
        new_notification_cv_.notify_all();
    }

    return ret_val;
}

SequenceNumber_t RTPSReader::get_last_notified(const GUID_t& guid) {
    SequenceNumber_t ret_val;
    std::lock_guard<RecursiveTimedMutex> guard(mp_mutex);
    GUID_t guid_to_look = guid;
    auto p_guid = history_state_->persistence_guid_map.find(guid);
    if (p_guid != history_state_->persistence_guid_map.end()) {
        guid_to_look = p_guid->second;
    }

    auto p_seq = history_state_->history_record.find(guid_to_look);
    if (p_seq != history_state_->history_record.end()) {
        ret_val = p_seq->second;
    }

    return ret_val;
}

void RTPSReader::set_last_notified(const GUID_t& peristence_guid, const SequenceNumber_t& seq) {
    history_state_->history_record[peristence_guid] = seq;
}

bool RTPSReader::wait_for_unread_cache(const evbs::ertps::Duration_t& timeout) {
    auto time_out = std::chrono::steady_clock::now() + std::chrono::seconds(timeout.seconds) +
                    std::chrono::nanoseconds(timeout.nanosec);

    std::unique_lock<RecursiveTimedMutex> lock(mp_mutex, std::defer_lock);

    if (lock.try_lock_until(time_out)) {
        if (new_notification_cv_.wait_until(lock, time_out, [&]() { return mp_history->total_unread_ > 0; })) {
            return true;
        }
    }

    return false;
}

uint64_t RTPSReader::get_unread_count() const {
    //std::unique_lock<RecursiveTimedMutex> lock(mp_mutex);
    return mp_history->total_unread_;
}

uint64_t RTPSReader::get_unread_count(bool mark_as_read) {
    uint64_t ret_val = mp_history->total_unread_;

    if (mark_as_read) {
        std::unique_lock<RecursiveTimedMutex> lock(mp_history->getMutex());
        for (auto it = mp_history->changesBegin(); (0U < mp_history->total_unread_) && (it != mp_history->changesEnd());
             ++it) {
            CacheChange_t* change = *it;
            if ((!change->isRead) && (get_last_notified(change->writerGUID) >= change->sequenceNumber)) {
                change->isRead = true;
                assert(0 < mp_history->total_unread_);
                --mp_history->total_unread_;
            }
        }
        assert(0 == mp_history->total_unread_);
    }
    return ret_val;
}

#ifdef BATCH_SEND_ENABLE
RTPS_DllAPI bool RTPSReader::parseChangeFromBatch(CacheChange_t* change_to_add, CacheChange_t* change_batch,
                                                  uint32_t& sample_pos, uint32_t& data_pos) {
    if (data_pos == 0U) {
        data_pos = RTPSMESSAGE_SAMPLE_INFO_SIZE * change_batch->batchSampleCount + RTPSMESSAGE_ENCAPSULATION_SIZE;
    }

    if ((data_pos >= change_batch->serializedPayload.length) || (data_pos <= sample_pos)) {
        elogError(RTPS_MSG_IN, RetCode_t::RETCODE_ERROR,
                  IDSTRING "batch buffer invalid.(" << sample_pos << "," << data_pos << ","
                                                    << change_batch->serializedPayload.length << ")");
        return false;
    }

    octet* sample_info = change_batch->serializedPayload.data + sample_pos;
    change_to_add->serializedPayload.length = *reinterpret_cast<uint32_t*>(&sample_info[4]);
    change_to_add->serializedPayload.data = change_batch->serializedPayload.data + data_pos;
    change_to_add->serializedPayload.max_size = change_to_add->serializedPayload.length;
    change_to_add->copy_not_memcpy(change_batch);

    sample_pos += RTPSMESSAGE_SAMPLE_INFO_SIZE;
    data_pos += change_to_add->serializedPayload.length;
    return true;
}
#endif

bool RTPSReader::is_sample_valid(const void* data, const GUID_t& writer,
                                 const SequenceNumber_t& sn) const {  //LCOV_EXCL_START
    UNUSED_PARAMETER(data);
    UNUSED_PARAMETER(writer);
    UNUSED_PARAMETER(sn);
    return true;
}  //LCOV_EXCL_STOP

std::shared_ptr<IPayloadPool> RTPSReader::get_payload_pool(ReaderHistory* hist) {
    // When the user requested PREALLOCATED_WITH_REALLOC, but we know the type cannot
    // grow, we translate the policy into bare PREALLOCATED
    PoolConfig config = PoolConfig::from_history_attributes(hist->m_att);

    std::shared_ptr<ITopicPayloadPool> payloadpool = TopicPayloadPoolRegistry::get(hist->topicName, config);
    (void)payloadpool->reserve_history(config, true);
    return payloadpool;
}

void RTPSReader::release_payload_pool() {
    assert(payload_pool_);

    PoolConfig config = PoolConfig::from_history_attributes(mp_history->m_att);
    payload_pool_->release_history(config, true);
    payload_pool_.reset();
}

void RTPSReader::add_announce_locator(Locator_t loc) {
    EndpointAttributes& attr = getAttributes();
    logDebug(RTPS_READER, "Reader add announce locator: " << loc);
    if (loc.kind == LOCATOR_PORT_INVALID) {
        logDebug(RTPS_READER, "locator is invalid");
        LocatorList_t locators = getRTPSParticipant()->getRTPSParticipantAttributes().defaultUnicastLocatorList;
        for (auto loc : locators) {
            attr.unicastLocatorList.push_back(loc);
        }
        return;
    }
    attr.unicastLocatorList.push_back(loc);
}

bool RTPSReader::history_adjustment() {
    mem_watermark += mp_history->getHistorySize();
    mem_watermark = mem_watermark / 2;
    mem_watermark = mem_watermark > 1 ? mem_watermark : 1;
    return payload_pool_->free_payload(mem_watermark);
}

} /* namespace rtps */
} /* namespace ertps */
} /* namespace evbs */
